package com.zhj.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zhj.kafka.entity.Order;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.math.BigDecimal;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * 　　　　　　　   ┏┓　   ┏┓+ +
 * 　　　　　　　┏┛┻━━━┛┻┓ + +
 * 　　　　　　　┃　　　　　　　┃
 * 　　　　　　　┃　　　━　　　┃ ++ + + +
 * 　　　　　　 ████━████ ┃+
 * 　　　　　　　┃　　　　　　　┃ +
 * 　　　　　　　┃　　　┻　　　┃
 * 　　　　　　　┃　　　　　　　┃ + +
 * 　　　　　　　┗━┓　　　┏━┛
 * 　　　　　　　　　┃　　　┃
 * 　　　　　　　　　┃　　　┃ + + + +
 * 　　　　　　　　　┃　　　┃			God beast body, code no BUG
 * 　　　　　　　　　┃　　　┃ +			神兽护体,代码无BUG
 * 　　　　　　　　　┃　　　┃
 * 　　　　　　　　　┃　　　┃　　+
 * 　　　　　　　　　┃　 　　┗━━━┓ + +
 * 　　　　　　　　　┃ 　　　　　　　┣┓
 * 　　　　　　　　　┃ 　　　　　　　┏┛
 * 　　　　　　　　　┗┓┓┏━┳┓┏┛ + + + +
 * 　　　　　　　　　　┃┫┫　┃┫┫
 * 　　　　　　　　　　┗┻┛　┗┻┛+ + + +
 *
 * @title:
 * @author zhonghaijun
 * @date 2021-04-15
 */
public class KafkaProducerApplication {

    public static final ObjectMapper objectMapper = new ObjectMapper();

    public static void main(String[] args) throws InterruptedException, JsonProcessingException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"47.116.130.99:21500");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG,"producer-1");
        /**
         * acks:
         *      0：异步不需要等到broker确认消息
         *      1：至少等待leader数据成功写入本地log
         *      -1或all：意味着leader需要等待所有备份，包括副本。强数据保证，影响性能
         */
        properties.put(ProducerConfig.ACKS_CONFIG,"1");
        //消息重试机制，可能会导致消息重复发送
        properties.put(ProducerConfig.RETRIES_CONFIG,3);
        //消息重试时间
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,300);
        //本地消息缓冲区，消息发送到本地缓冲区，可提高发送性能，默认32m
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
        //kafka先从缓冲区中取出消息，进行批量发送。默认16kb
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
        //如果100毫秒内，batch没满也要发送
        properties.put(ProducerConfig.LINGER_MS_CONFIG,100);
        //把发送的key从字符串序列化为字节数组
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送的value从字符串序列化为字节数组
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String,String>(properties);
        int msgNum = 5;
        CountDownLatch countDownLatch = new CountDownLatch(msgNum);
        for (int i = 0; i < msgNum; i++) {
            Order order = new Order(i,100+i,new BigDecimal(1000));
            //发送消息。如果未指定发送分区，分区的公式  hash(key)%partitionNum
            //asyncMethod(kafkaProducer, order,countDownLatch);
            syncMethod(kafkaProducer,order,countDownLatch);
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        kafkaProducer.close();
    }
    //同步方式发送消息
    private static void asyncMethod(KafkaProducer<String, String> kafkaProducer, Order order,CountDownLatch countDownLatch) throws InterruptedException, ExecutionException, JsonProcessingException {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test",
                                                                             order.getKeyID().toString(),
                                                                             objectMapper.writeValueAsString(order));
        RecordMetadata recordMetadata = kafkaProducer.send(producerRecord).get();
        System.out.println("同步发送消息结果：topic"+recordMetadata.topic()+" partition："+recordMetadata.partition()+" offset:"+recordMetadata.offset());
        countDownLatch.countDown();
    }
    //异步方式发送
    private static void syncMethod(KafkaProducer<String, String> kafkaProducer, Order order,CountDownLatch countDownLatch) throws JsonProcessingException {
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test-1",
                                                                             order.getKeyID().toString(),
                                                                             objectMapper.writeValueAsString(order));
        kafkaProducer.send(producerRecord, (recordMetadata, e) -> {
            if(e != null){
                System.out.println("消息发送失败：" + Arrays.toString(e.getStackTrace()));
            }
            System.out.println("异步发送消息结果：topic"+recordMetadata.topic()+" partition："+recordMetadata.partition()+" offset:"+recordMetadata.offset());
            countDownLatch.countDown();
        });
    }

}
